package SparkStreaming;
/**
 * supplied 提供的
 * varying 变化
 * multiples 背书
 * discard  抛弃 放弃
 */


import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.xalan.lib.sql.ConnectionPool;
import scala.Tuple2;

import java.util.Arrays;

/**
 * @program: MySpark
 * @description
 * @author: tkk fendoukaoziji
 * @create: 2019-04-23 16:38
 **/
public class WindowOperations {
    public static void main(String[] args) {
        //Logger.getLogger("org.apache.spark").setLevel(Level.WARN); //设置日志级别warn
        SparkConf conf = new SparkConf().setAppName("repeatSparkStreaming").setMaster("local[*]");
        JavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.seconds(5));
        ssc.checkpoint("text");
        JavaReceiverInputDStream<String> lines = ssc.socketTextStream("node3", 1234);
        JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());
        JavaPairDStream<String, Integer> pairs= words.mapToPair(x -> new Tuple2<>(x, 1));
        JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKeyAndWindow((v1, v2) ->
            v1 + v2, Durations.seconds(20), Durations.seconds(10));
        wordCounts.print();
        ssc.start();
        try {
            ssc.awaitTermination();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
